﻿// Copyright (c) .NET Core Community. All rights reserved.
// Licensed under the MIT License. See License.txt in the project root for license information.

using DotNetCore.CAP;
using DotNetCore.CAP.Internal;
using DotNetCore.CAP.Messages;
using DotNetCore.CAP.RabbitMQ;
using DotNetCore.CAP.Transport;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using RabbitMQ.Client;

namespace Seagull.Extensions.ServiceBus
{
    internal sealed class RabbitMQTransport : ITransport
    {
        private readonly IConnectionChannelPool _connectionChannelPool;
        private readonly ILogger _logger;
        private readonly string _exchange;
        private readonly DelayOptions _options;

        public RabbitMQTransport(
            ILogger<RabbitMQTransport> logger,
            IConnectionChannelPool connectionChannelPool,
            IOptions<DelayOptions> options)
        {
            _logger = logger;
            _connectionChannelPool = connectionChannelPool;
            _exchange = _connectionChannelPool.Exchange;
            _options = options.Value;
        }

        public BrokerAddress BrokerAddress => new BrokerAddress("RabbitMQ", _connectionChannelPool.HostAddress);

        public Task<OperateResult> SendAsync(TransportMessage message)
        {
            IModel channel = null;
            try
            {
                channel = _connectionChannelPool.Rent();

                channel.ConfirmSelect();

                var props = channel.CreateBasicProperties();
                props.DeliveryMode = 2;
                props.Headers = message.Headers.ToDictionary(x => x.Key, x => (object)x.Value);

                //判断是否是延时消息
                if (props.Headers.ContainsKey("x-delay"))
                {
                    channel.ExchangeDeclare(_options.ExchangeName, "x-delayed-message", true);
                    channel.BasicPublish(_options.ExchangeName, message.GetName(), props, message.Body);
                }
                else
                {
                    channel.ExchangeDeclare(_exchange, RabbitMQOptions.ExchangeType, true);

                    channel.BasicPublish(_exchange, message.GetName(), props, message.Body);
                }


                channel.WaitForConfirmsOrDie(TimeSpan.FromSeconds(5));

                _logger.LogInformation("CAP message '{0}' published, internal id '{1}'", message.GetName(), message.GetId());

                return Task.FromResult(OperateResult.Success);
            }
            catch (Exception ex)
            {
                var wrapperEx = new PublisherSentFailedException(ex.Message, ex);
                var errors = new OperateError
                {
                    Code = ex.HResult.ToString(),
                    Description = ex.Message
                };

                return Task.FromResult(OperateResult.Failed(wrapperEx, errors));
            }
            finally
            {
                if (channel != null)
                {
                    _connectionChannelPool.Return(channel);
                }
            }
        }
    }
}